package event

import (
	"context"
	"errors"
	"net"
	"reflect"
	"sieve_admin_server_userpass/pkg/app"
	"sieve_admin_server_userpass/pkg/config"
	"sieve_admin_server_userpass/pkg/im3/crypto"
	"sieve_admin_server_userpass/pkg/im3/event/cli"
	"sieve_admin_server_userpass/pkg/json"
	"sieve_admin_server_userpass/pkg/log"
	"sieve_admin_server_userpass/pkg/xconn"
	"strconv"
	"strings"
	"time"
)

var (
	fServer = config.String("event.server", "im.manyji.net:9998", "event server")
	client  = &evClient{subscibe: map[string]reflect.Type{}}
)

func init() {
	app.OnStart(func() {
		app.Async(func() {
			for {
				if err := client.Connect(); err != nil {
					log.Err(err)
					time.Sleep(time.Second)
					continue
				}
				select {
				case <-app.CTX.Done():
					break
				case <-client.ctx.Done():
					log.Info("event connection closed")
					continue
				}
			}
		})
	})
}

type loginData struct {
	Name      string // im-server,svc-xxx,app-123
	Subscibe  []byte // aes(subscibe)
	LastMsgId int64
}

type evClient struct {
	conn      *xconn.Conn
	subscibe  map[string]reflect.Type
	ctx       context.Context
	cancel    context.CancelFunc
	lastMsgId int64
}

func (ec *evClient) Connect() error {
	conn, err := net.Dial("tcp", *fServer)
	if err != nil {
		return err
	}
	ec.ctx, ec.cancel = context.WithCancel(app.CTX)
	ec.conn, err = xconn.NewConn(conn, &xconn.Config{OnInit: func(rw xconn.RawRW) error {
		if cli.Name == "" || len(cli.Pwd) != 32 {
			return errors.New("event client require init")
		}
		connLog := "connect to event server " + *fServer
		if ec.lastMsgId > 0 {
			connLog += " fetch since " + strconv.FormatInt(ec.lastMsgId, 10)
		}
		subscibes := []string{}
		for s := range ec.subscibe {
			subscibes = append(subscibes, s)
		}
		j, err := json.Marshal(subscibes)
		if err != nil {
			return err
		}
		a, err := crypto.AesEncrypt_bytes(cli.Pwd, j)
		if err != nil {
			return err
		}
		b, err := json.Marshal(&loginData{Name: cli.Name, Subscibe: a, LastMsgId: ec.lastMsgId})
		if err != nil {
			return err
		}
		if ec.lastMsgId == 0 {
			ec.lastMsgId = time.Now().UnixNano()
		}
		return rw.Write(b)
	}, OnClosed: func() {
		ec.cancel()
	}})
	if err == nil {
		ec.conn.Handle(ec.HandleMsg)
	}
	return err
}

type eventMsg struct {
	Id   int64
	From string
	Name string
	Data json.RawMessage
}

func (ec *evClient) HandleMsg(b []byte) []byte {
	msg := &eventMsg{}
	if err := json.Unmarshal(b, &msg); err != nil {
		log.Err(err)
		return nil
	}
	ec.lastMsgId = msg.Id
	t, ok := ec.subscibe[msg.Name]
	if !ok {
		log.Err(msg.Name, "not found")
		return nil
	}
	v := reflect.New(t).Interface()
	if err := json.Unmarshal(msg.Data, &v); err != nil {
		log.Err(err)
		return nil
	}
	v.(Event).Handle(msg.From)
	return nil
}

type Event interface {
	Handle(from string)
}

// 在init()中调用
// type event_消息名
func Listen(ev Event) error {
	t := reflect.TypeOf(ev).Elem()
	name := t.Name()
	if !strings.HasPrefix(name, "event_") {
		return errors.New("require prefix event_")
	}
	client.subscibe[name[6:]] = t
	return nil
}

// type event_消息名
func Dispatch(ev interface{}) error {
	t := reflect.TypeOf(ev).Elem()
	name := t.Name()
	if !strings.HasPrefix(name, "event_") {
		return errors.New("require prefix event_")
	}
	b, err := json.Marshal(ev)
	if err != nil {
		return err
	}
	msg := &eventMsg{From: cli.Name, Name: name[6:], Data: json.RawMessage(b)}
	j, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	if client.conn == nil {
		return errors.New("connection not available")
	}
	return client.conn.Post(j)
}
